package mapreduce.phase2;

import java.io.IOException;
import java.util.StringTokenizer;

import mapreduce.Point;
import mapreduce.ColumnGrp;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;


/**
 * Get the global picture of the shared points
 * skip -1  & -2
 * Map: only pass the points in boundaries: 
 */

@SuppressWarnings("deprecation")
public class Map2 extends MapReduceBase implements
	Mapper<LongWritable, Text, IntWritable, Text> {
		private final static IntWritable part = new IntWritable(1);

		public void map(LongWritable key, Text value,
				OutputCollector<IntWritable, Text> output, Reporter reporter)
		throws IOException {
			String line = value.toString();
			StringTokenizer tokenizer = new StringTokenizer(line);
			assert (tokenizer.hasMoreTokens());
			int grp = Integer.parseInt(tokenizer.nextToken());
			if(grp<0)
				return;
			assert (tokenizer.hasMoreTokens());
			Point p = Point.fromString(tokenizer.nextToken());
			assert (tokenizer.hasMoreTokens());
			Point q = Point.fromString(tokenizer.nextToken());
			int secondaryGroup = ColumnGrp.getSecondryGroup(p);
			if(secondaryGroup != -1) {
				Text t = new Text(p.toString() + " " + q.toString());
				part.set(1);
				output.collect(part, t);
			}
		}
	}
